package cn.panshi.mongocdc;

import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;

import org.bson.Document;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.messaging.Message;
import org.springframework.data.redis.core.RedisTemplate;

import com.mongodb.client.model.changestream.ChangeStreamDocument;
import com.mongodb.client.model.changestream.OperationType;

import lombok.extern.slf4j.Slf4j;

@Slf4j
@SuppressWarnings("rawtypes")
public abstract class AbsCdcTemplate4CreateUpdateRouter extends AbsCdcTemplate {

	public AbsCdcTemplate4CreateUpdateRouter(RedisTemplate redisTemplate, MongoTemplate mongoTemplate) {
		super(redisTemplate, mongoTemplate);
	}

	@Override
	protected boolean supportsOperation(String collectionName, OperationType operationType) {
		List<Enum> cdcTypes = this.parseCdcType(collectionName, operationType);
		if (cdcTypes == null || cdcTypes.isEmpty()) {
			log.trace("No cdcType matched 4 cdc:{} {}", collectionName, operationType.name());
			return false;
		}
		List<Method> matchedBizList = new ArrayList<>();
		for (Enum cdcType : cdcTypes) {
			List<Method> matchedBiz = this.matchBizMethodByCdc(cdcType);
			if (matchedBiz == null || matchedBiz.isEmpty()) {
				log.debug("No biz metched 4 {} {} {}", cdcType.getClass().getName(), cdcType.name(),
						operationType.name());
				return false;
			}
			matchedBizList.addAll(matchedBiz);
		}

		return matchedBizList.size() > 0;
	}

	@Override
	protected void executeCdcBiz(String bizId, String resumeToken,
			Message<ChangeStreamDocument<Document>, Document> message, String collectionName,
			OperationType operationType) throws Exception {

		List<Enum> cdcTypes = this.parseCdcType(collectionName, operationType);
		for (Enum cdcType : cdcTypes) {
			List<Method> matchedRouterFuncs = this.matchBizMethodByCdc(cdcType);

			for (Method m : matchedRouterFuncs) {
				this.invokeBizRouter(m, bizId, resumeToken, message);
			}
		}
	}

	abstract protected List<Method> matchBizMethodByCdc(Enum cdcType);

	abstract protected List<Enum> parseCdcType(String collectionName, OperationType operationType);

	abstract protected void invokeBizRouter(Method matchedRouterFunc, String bizId, String resumeToken,
			Message<ChangeStreamDocument<Document>, Document> message) throws Exception;
}